package t20240717;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.types.RowKind;

import java.sql.Timestamp;
import java.util.ArrayList;

import static org.apache.flink.table.api.Expressions.row;

public class Tum {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);



        Table table = tEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("stime", DataTypes.TIMESTAMP()),
                        DataTypes.FIELD("name", DataTypes.STRING()),
                        DataTypes.FIELD("val", DataTypes.INT())
                ),
                row(new Timestamp(2024, 7, 17, 12, 0, 0,  0), "AA", 1),
                row(new Timestamp(2024, 7, 17, 12, 1, 0,  0), "BB", 2),
                row(new Timestamp(2024, 7, 17, 12, 2, 0,  0), "AA", 3),
                row(new Timestamp(2024, 7, 17, 12, 3, 0,  0), "BB", 4),
                row(new Timestamp(2024, 7, 17, 12, 4, 0,  0), "AA", 5)
        );

        tEnv.registerTable("ods_tb", table);

        String sql = "CREATE VIEW v_base AS\n" +
                "SELECT * FROM (\n" +
                "SELECT\n" +
                "stime,\n" +
                "    name," +
                "        val,\n" +
                //"        ROW_NUMBER() OVER (PARTITION BY name ORDER BY stime DESC) AS rownum " +
                //"        ROW_NUMBER() OVER (PARTITION BY name ORDER BY stime DESC RANGE BETWEEN INTERVAL '1' MINUTE PRECEDING AND CURRENT ROW) AS rownum " +
                "        ROW_NUMBER() OVER (PARTITION BY TUMBLE(stime, INTERVAL '1' MINUTE),name ORDER BY stime DESC) as rownum " +
                "    from ods_tb \n" +
                ")WHERE rownum = 1";

        System.out.println(sql);

        tEnv.executeSql(sql);


        tEnv.executeSql("select * from v_base").print();



    }
}
